package javacourse.mq.mymq.service;

import javacourse.mq.mymq.Queue.MyQueue;
import javacourse.mq.mymq.Queue.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * Create by chenhd on 2021/11/6.
 */
@Component
public class QueueOperator {

    @Autowired
    MyQueue myQueue;

    /**
     * 生产消息
     *
     * @param topic
     * @param message
     * @return
     */
    public boolean produce(String topic, String message) {
        try {
            //获取当前topic消息应发往的partition，目前采用轮询
            Integer partition = myQueue.getNextPartition(topic);
            TopicPartition topicPartition = new TopicPartition(topic, partition);
            //由于是个无界数组，这里不判断数组满的情况
            List<String> messageList = myQueue.getQueue(topicPartition);
            messageList.add(message);


            //唤醒阻塞住的消费者
            //这里使用了锁，当然在高并发场景下，这不是一个很好的解决办法，以下消费者逻辑同
            //可以考虑的改进点：更细粒度的锁，无锁的设计
            Lock lock = myQueue.getMessageLock();
            try {
                lock.lock();
                //产生新消息，通知可以消费了
                Condition condition = myQueue.getConditionMap().get(topicPartition);
                condition.signalAll();
            } catch (Exception e) {
                e.printStackTrace();
                //log
            } finally {
                lock.unlock();
            }

            return true;
        } catch (Exception e) {
            //log
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 消费消息，如果消息队列为空则等待timeout秒
     *
     * @param topic
     * @param consumer
     * @param timeout  消费超时时间，单位秒
     * @return
     */
    public Map<String, Object> consume(String topic, String consumer, long timeout) {
        Lock lock = myQueue.getMessageLock();
        try {
            //要考虑线程安全
            lock.lock();
            Map<String, Object> resultMap = new HashMap<>();
            Map<TopicPartition, String> partitionConsumerMap = myQueue.getPartitionConsumerMap();

            for (int partition = 0; partition < MyQueue.PARTITIONS; partition++) {
                TopicPartition topicPartition = new TopicPartition(topic, partition);
                //找到consumer对应的topicPartition
                if (Objects.equals(partitionConsumerMap.get(topicPartition), consumer)) {
                    List<String> messageQueue = myQueue.getQueue(topicPartition);
                    //边界值判定，借助Lock Condition机制
                    while (messageQueue.size() <= myQueue.getOffset(topicPartition)) {
                        //如果超时，则直接返回
                        if (!myQueue.getConditionMap().get(topicPartition).await(timeout, TimeUnit.SECONDS)) {
                            return null;
                        }
                    }
                    int offset = myQueue.getOffset(topicPartition);
                    String message = messageQueue.get(offset);
                    resultMap.put("message", message);
                    resultMap.put("offset", offset + 1);
                }
            }
            return resultMap;
        } catch (Exception e) {
            //log
            e.printStackTrace();
            return null;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 提交offset
     *
     * @param topic
     * @param consumer
     * @param offset
     * @return
     */
    public boolean commitOffset(String topic, String consumer, Integer offset) {
        try {
            for (int partition = 0; partition < MyQueue.PARTITIONS; partition++) {
                TopicPartition topicPartition = new TopicPartition(topic, partition);
                Map<TopicPartition, String> partitionConsumerMap = myQueue.getPartitionConsumerMap();
                if (Objects.equals(partitionConsumerMap.get(topicPartition), consumer)) {
                    //最大offset不会超过队列长度
                    myQueue.setOffset(topicPartition, Math.min(offset, myQueue.getQueue(topicPartition).size() - 1));
                    return true;
                }
            }
        } catch (Exception e) {
            //log
            e.printStackTrace();
        }
        return false;
    }

    public boolean assignConsumer(String topic, List<String> consumers) {
        try {
            myQueue.assignPartition(topic, consumers);
            return true;
        } catch (Exception e) {
            //log
            e.printStackTrace();
            return false;
        }
    }
}
